package com.example.messagequeue.mqserver.core;

import com.example.messagequeue.commom.Consumer;
import com.example.messagequeue.commom.ConsumerEnv;
import com.example.messagequeue.commom.MqException;
import com.example.messagequeue.mqserver.VirtualHost;

import java.util.concurrent.*;

/*
 * t通过这个类来实现消费消息的核心逻辑
 */
public class ConsumerManager {
    //持有上层的VirtualHost对象的引用，用来操作数据
    private VirtualHost parent;
    //指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPoor = Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;

        scannerThread = new Thread(() -> {
            while (true){
                try {
                    //拿到令牌
                    String queueName = tokenQueue.take();
                    //根据令牌找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null){
                        throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在! queueName = " + queueName);
                    }
                    synchronized (queue){
                        consumerMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        scannerThread.setDaemon(true);
        scannerThread.start();

    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null){
            throw new MqException("[ConsumerManager] 队列不存在！ queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++){
                //调用一次消费一条消息
                consumerMessage(queue);
            }
        }
    }

    private void consumerMessage(MSGQueue queue) {
        //1.按照轮询的方式，找个消费者出来
        ConsumerEnv luckDog = queue.chooseConsumer();
        if (luckDog == null){
            //当前没有消费者，暂时不消费,等后面有消费者出现再说
            return;
        }
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null){
            return;
        }
        workerPoor.submit(() -> {
            try {
                //1.把消息放到待确认的集合中，这个操作必须得在执行回调函数之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2.执行回调函数
                luckDog.getConsumer().handleDelivery(luckDog.getConsumerTag(), message.getBasicProperties(), message.getBody());

                //3.如果是自动应答，就可以直接把消息删除
                //如果是手动应答，则先不处理，交给后续消费者使用basicAck方法来处理
                if (luckDog.isAutoAck()){
                    if (message.getDeliverMode() == 2){
                        parent.getDisDataCenter().deleteMessage(queue,message);
                    }
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName = " + queue.getName());
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }
}
